#ifndef __M__BROKERSERVER_H_
#define __M__BROKERSERVER_H_

// 网络通信相关头文件
#include "../third/muduo/include/muduo/net/TcpServer.h"
#include "../third/muduo/include/muduo/net/EventLoop.h"
#include "../third/muduo/include/muduo/net/TcpConnection.h"
#include "../third/muduo/include/muduo/net/Buffer.h"
// 协议处理和分发
#include "../third/muduo/proto/dispatcher.h"
#include "../third/muduo/proto/codec.h"
// 自定义应用层协议（protobuf）
#include "../mqCommon/proto/mq_message.pb.h"
#include "../mqCommon/proto/mq_proto.pb.h"

// 各模块
#include "virtual_host.hpp"
#include "consumer.hpp"
#include "../mqCommon/threadPool.hpp"
#include "connection.hpp"
#include "../mqCommon/Logger.hpp"

namespace mq
{
    class Server
    {
    public:
        Server(int port, const string &basedir, const string &dbfile = "server.db")
            : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort),
              _dispatcher(std::bind(&Server::onUnknownMessage, this, placeholders::_1, placeholders::_2, placeholders::_3)),
              _codec(make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, placeholders::_1, placeholders::_2, placeholders::_3))),
              _virtual_host(make_shared<VirtualHost>("host", basedir, dbfile)),
              _consumer_manager(make_shared<ConsumerManager>()),
              _connection_manager(make_shared<ConnectionManager>()),
              _thread_pool(make_shared<ThreadPool>())
        {
            // 1.实例化各管理模块:恢复历史信息
            // 获取所有队列，初始化队列消费者
            MsgQueueMap qmap = _virtual_host->allQueues();
            for (auto &it : qmap)
            {
                _consumer_manager->initQueueConsumer(it.first);
            }

            // 2.注册请求处理回调
            _dispatcher.registerMessageCallback<openChannelRequest>(std::bind(&Server::onOpenChannel, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<closeChannelRequest>(std::bind(&Server::onCloseChannel, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<declareExchangeRequest>(std::bind(&Server::onDeclareExchange, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<deleteExchangeRequest>(std::bind(&Server::onDeleteExchange, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<declareQueueRequest>(std::bind(&Server::onDeclareQueue, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<deleteQueueRequest>(std::bind(&Server::onDeleteQueue, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<queueBindRequest>(std::bind(&Server::onBind, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<queueUnBindRequest>(std::bind(&Server::onUnBind, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<basicPublishRequest>(std::bind(&Server::onPublish, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<basicAckRequest>(std::bind(&Server::onAck, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeRequest>(std::bind(&Server::onConsume, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<basicCancelRequest>(std::bind(&Server::onCancel, this, placeholders::_1, placeholders::_2, placeholders::_3));
            // 绑定消息回调函数
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                                                 placeholders::_1, placeholders::_2, placeholders::_3));
            // 绑定连接回调函数
            _server.setConnectionCallback(std::bind(&Server::onConnection, this, placeholders::_1));
        }

    private:
        // 向请求分发器注册请求处理回调：
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp) // 收到未定义的消息类型
        {
            cout << "onUnknownMessage:" << message->GetTypeName() << endl;
        }
        void onOpenChannel(const TcpConnectionPtr &conn, const openChannelRequestPtr &req, muduo::Timestamp) // 打开信道请求
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                DLOG("关闭连接。。。");
                return;
            }
            Connection::ptr connp = _connection_manager->getConnection(conn);

            connp->openChannel(req); // 打开对应连接信道
        }
        void onCloseChannel(const TcpConnectionPtr &conn, const closeChannelRequestPtr &req, muduo::Timestamp) // 关闭信道请求
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            Connection::ptr connp = _connection_manager->getConnection(conn);

            connp->closeChannel(req); // 关闭对应连接信道
        }
        void onDeclareExchange(const TcpConnectionPtr &conn, const declareExchangeRequestPtr &req, muduo::Timestamp) // 声明交换机
        {
            // 根据连接，找到对应信道，由信道提供服务
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.声明交换机
            channel->declareExchange(req);
        }
        void onDeleteExchange(const TcpConnectionPtr &conn, const deleteExchangeRequestPtr &req, muduo::Timestamp) // 删除交换机
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.删除交换机
            channel->deleteExchange(req);
        }
        void onDeclareQueue(const TcpConnectionPtr &conn, const declareQueueRequestPtr &req, muduo::Timestamp) // 声明队列
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.声明队列
            channel->declareMsgQueue(req);
        }
        void onDeleteQueue(const TcpConnectionPtr &conn, const deleteQueueRequestPtr &req, muduo::Timestamp) // 删除队列
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.删除队列
            channel->deleteMsgQueue(req);
        }
        void onBind(const TcpConnectionPtr &conn, const queueBindRequestPtr &req, muduo::Timestamp) // 绑定
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.绑定
            channel->bind(req);
        }
        void onUnBind(const TcpConnectionPtr &conn, const queueUnBindRequestPtr &req, muduo::Timestamp) // 解除绑定
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.解绑定
            channel->unBind(req);
        }
        void onPublish(const TcpConnectionPtr &conn, const basicPublishRequestPtr &req, muduo::Timestamp) // 发布消息
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.发布消息
            channel->basicPublish(req);
        }
        void onAck(const TcpConnectionPtr &conn, const basicAckRequestPtr &req, muduo::Timestamp) // 确认消息
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.消息确认
            channel->basicAck(req);
        }
        void onConsume(const TcpConnectionPtr &conn, const basicConsumeRequestPtr &req, muduo::Timestamp) // 订阅队列
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.订阅队列
            channel->basicConsume(req);
        }
        void onCancel(const TcpConnectionPtr &conn, const basicCancelRequestPtr &req, muduo::Timestamp) // 解除订阅
        {
            if (!conn_exist(conn))
            {
                conn->shutdown();
                return;
            }
            // 1.获取连接
            Connection::ptr connp = _connection_manager->getConnection(conn);
            // 2.获取信道
            Channel::ptr channel = connp->getChannel(req->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到对应信道：%s", req->cid());
                return;
            }
            // 3.取消订阅
            channel->basicCancel(req);
        }
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                // 创建连接
                _connection_manager->newConnection(_consumer_manager, conn, _codec, _virtual_host, _thread_pool);
                cout << "新连接建立成功" << endl;
            }
            else
            {
                // 关闭连接
                _connection_manager->delConnection(conn);
                cout << "连接关闭" << endl;
            }
        }
        bool conn_exist(const TcpConnectionPtr &tcp_conn)
        {
            Connection::ptr connp = _connection_manager->getConnection(tcp_conn);
            if (connp.get() == nullptr)
            {
                DLOG("关闭信道时，未找到对应连接：%s", tcp_conn->name());
                return false;
            }
            return true;
        }

    public:
        void Start()
        {
            _server.start();
            _baseloop.loop();
        }

    private:
        muduo::net::EventLoop _baseloop; // 事件监控
        muduo::net::TcpServer _server;   // 服务器
        ProtobufDispatcher _dispatcher;  // 请求分发器：注册请求处理函数
        ProtobufCodecPtr _codec;         // 协议处理器：针对字节流数据协议处理（len|typeLen|typeName|protobufData|checkSum）

        VirtualHost::ptr _virtual_host;             // 虚拟机：核心数据管理（交换机/队列/绑定/消息）
        ConsumerManager::ptr _consumer_manager;     // 消费者管理（队列订阅者）
        ConnectionManager::ptr _connection_manager; // 连接（信道）管理
        ThreadPool::ptr _thread_pool;               // 异步工作线程池（异步消息推送任务）
    };
}

#endif
